package chen.bupt.mapreduce.board.semantic;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import chen.bupt.constant.Constants;
import chen.bupt.text.TextClear;
import chen.bupt.text.TextUtils;
import chen.bupt.util.HDFSFileUtils;

/**
 * 寻找词对，根据共现次数，用于识别语义相近词汇
 * 实验效果不好，不采用
 * 
 * @author forrest
 * 
 */
public class SynonymsJob {
	private static final String inputBase1 = Constants.BOARD_DATA;
	private static final String outputBase1 = "/user/chenlingpeng/smth/board_synonyms/count/";

	private static final String outputBase2 = "/user/chenlingpeng/smth/board_synonyms/final/";
	// 根据得分进行完整排序
	private static final String outputBase3 = "/user/chenlingpeng/smth/board_synonyms/final_sort/";

	public static class SynonymsMap extends
			Mapper<LongWritable, Text, Text, IntWritable> {
		private static final IntWritable one = new IntWritable(1);

		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			if (!HDFSFileUtils.isInputFile(context, "-r-"))
				return;
			// TODO 清理前缀无效项
			String tmp = TextUtils.splitTextToCont(value.toString(), ";", 6)[1];
			if (tmp == null || tmp.equals(""))
				return;
			String[] terms = tmp.split(" ");
			int len = terms.length;
			if (len < 2)
				return;
			for (int i = 0; i < len; i++) {
				if (terms[i].contains("/nr")) {
					terms[i] = TextClear.substring(terms[i], "/nr");
				}
			}
			for (int i = 0; i < len - 1; i++) {
				for (int j = i + 1; j < len; j++) {
					if (terms[i].equals(terms[j]))
						continue;
					if (terms[i].compareTo(terms[j]) > 0) {
						context.write(new Text(terms[i] + "_" + terms[j]), one);
					} else {
						context.write(new Text(terms[j] + "_" + terms[i]), one);
					}
				}
			}
		}
	}

	public static class SynonymsReduce extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		@Override
		protected void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			int cnt = 0;
			for (IntWritable val : values) {
				cnt += val.get();
			}
			context.write(key, new IntWritable(cnt));
		}
	}

	public static class SynonymsMap2 extends
			Mapper<LongWritable, Text, Text, Text> {

		private Random random;
		private final int range = 10000;

		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			// TODO Auto-generated method stub
			random = new Random();
		}

		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			if (!HDFSFileUtils.isInputFile(context))
				return;
			int fakeKey = random.nextInt() % range;
			context.write(new Text(fakeKey + ""), value);
		}
	}

	/**
	 * 两个词的共现程度
	 * 
	 * @author forrest
	 * 
	 */
	private static class PairScore {
		public String term1, term2;
		double score;

		public PairScore(String term1, String term2, double score) {
			this.term1 = term1;
			this.term2 = term2;
			this.score = score;
		}

		@Override
		public String toString() {
			return term1 + "," + term2 + "," + score;
		}

		public static PairScore parsePairScore(String str) {
			String[] s = str.split(",");
			return new PairScore(s[0], s[1], Double.parseDouble(s[2]));
		}

	}

	private static class MyCompatator implements Comparator<PairScore> {

		@Override
		public int compare(PairScore arg0, PairScore arg1) {
			// TODO Auto-generated method stub
			return arg0.score - arg1.score > 0 ? 1 : -1;
		}
	}

	// similary(A,B) = z(A,B)/SQRT(XA*XB)
	public static class SynonymsReduce2 extends
			Reducer<Text, Text, Text, NullWritable> {
		private static final String boardFeatureBase = "/user/chenlingpeng/smth/board_feature/";
		private static final String fileName = "/part-r-00000";
		private Map<String, Integer> featureMap = new HashMap<String, Integer>();

		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			String boardId = context.getConfiguration().get("boardId");
			String absluteFileName = boardFeatureBase + boardId + fileName;
			String feature = HDFSFileUtils.readLines(absluteFileName).get(0);
			String[] features = feature.split(";");
			for (String tmp : features) {
				if (null == tmp || "".equals(tmp))
					continue;
				featureMap.put(tmp.split(":")[0], Integer.parseInt(tmp
						.split(":")[1]));
			}
		}

		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			List<PairScore> list = new ArrayList<PairScore>();
			for (Text val : values) {
				String[] terms = val.toString().split("\t")[0].split("_");
				int cnt = Integer.parseInt(val.toString().split("\t")[1]);
				Integer t1 = featureMap.get(terms[0]);
				Integer t2 = featureMap.get(terms[1]);
				if (t1 == null || t2 == null)
					continue;
				double score = cnt / (Math.sqrt((double) t1 * t2));
				list.add(new PairScore(terms[0], terms[1], score));
			}
			Collections.sort(list, new MyCompatator());
			for (PairScore ps : list) {
				context.write(new Text(ps.toString()), NullWritable.get());
			}
		}
	}

	public static class SynonymsMap3 extends
			Mapper<LongWritable, Text, Text, Text> {
		private static Text fakeKey = new Text("one");

		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			context.write(fakeKey, value);
		}
	}

	public static class SynonymsReduce3 extends
			Reducer<Text, Text, Text, NullWritable> {

		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			List<PairScore> list = new ArrayList<PairScore>();
			for (Text val : values) {
				list.add(PairScore.parsePairScore(val.toString()));
			}
			Collections.sort(list, new MyCompatator());
			for (PairScore ps : list) {
				context.write(new Text(ps.toString()), NullWritable.get());
			}
		}

	}

	/**
	 * @param args
	 *            要进行job计算的boardId的前缀
	 */
	public static void main(String[] args) throws Exception {
		Configuration confg = new Configuration();
		confg.set("fs.default.name", "hdfs://10.1.1.26:31054");
		FileSystem fs = FileSystem.get(confg);
		FileStatus[] fsDirs = fs.listStatus(new Path(inputBase1));
		for (FileStatus fss : fsDirs) {
			if (!fss.isDir())
				continue;
			Path dirName = fss.getPath();
			String boardId = dirName.getName();
			if (boardId.startsWith("_"))
				continue;
			if (!boardId.startsWith(args[0]))
				continue;
			// 第一个job，统计共现次数
			Configuration conf = new Configuration();
			Job job = new Job(conf, "SynonymsJob1");
			job.setJarByClass(SynonymsJob.class);
			job.setMapperClass(SynonymsMap.class);
			job.setCombinerClass(SynonymsReduce.class);
			job.setReducerClass(SynonymsReduce.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(IntWritable.class);
			Path input = new Path(inputBase1 + "/" + boardId);
			Path output1 = new Path(outputBase1 + boardId);
			Path output2 = new Path(outputBase2 + boardId);
			HDFSFileUtils.deleteFile(output1, conf);
			HDFSFileUtils.deleteFile(output2, conf);
			FileInputFormat.addInputPath(job, input);
			FileOutputFormat.setOutputPath(job, output1);
			job.waitForCompletion(true);

			Configuration conf2 = new Configuration();
			conf2.set("boardId", boardId);
			Job job2 = new Job(conf2, "SynonymsJob2");
			job2.setJarByClass(SynonymsJob.class);
			job2.setMapperClass(SynonymsMap2.class);
			job2.setReducerClass(SynonymsReduce2.class);
			job2.setOutputKeyClass(Text.class);
			job2.setOutputValueClass(Text.class);
			FileInputFormat.addInputPath(job2, output1);
			FileOutputFormat.setOutputPath(job2, output2);
			job2.waitForCompletion(true);
		}

	}

}
